﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Runtime.CompilerServices;

using log4net;
using NHibernate;
using NHibernate.Criterion;

using Pink.Scheduler.DataAccess;
using Pink.Scheduler.Shared;
using Pink.Scheduler.RemoteInterfaces.Enum;

namespace Pink.Scheduler.RemoteInterfaces
{
	/// <summary>
	/// Диспетчер задач
	/// </summary>
	public class TaskDispatcher
	{
		private ILog logger = LogManager.GetLogger(typeof(TaskDispatcher));
		private ISessionFactory sessionFactory;
		/// <summary>
		/// Словарь "идентификатор задачи -- процессор"
		/// </summary>
		private Dictionary<string, TaskProcessor> processors;
		private ExecutionStateManager stateManager;
		private ITaskExecutor executor;
		private Schedule schedule;
		private Timer tasksUpdateTimer;
		private TimeSpan tasksUpdatePeriod;
		private int processingDelaySeconds;
		private Random generator = new Random();
		/// <summary>
		/// Конструктор
		/// </summary>
		/// <param name="factory">Фабрика сессий</param>
		/// <param name="executor">Кто занимается самим выполнением</param>
		/// <param name="processingDelaySeconds">Интервал для случайной задержки
		/// перед началом обработки задач</param>
		public TaskDispatcher(ISessionFactory factory, ITaskExecutor executor, 
			int processingDelaySeconds)
		{
			this.sessionFactory = factory;
			this.executor = executor;
			this.schedule = new Schedule(sessionFactory);
			this.processingDelaySeconds = processingDelaySeconds;
			stateManager = new ExecutionStateManager(factory);
		}

		/// <summary>
		/// Запустить выполнение задач
		/// </summary>
		/// <param name="tasksUpdatePeriod">период обновления состояния задач</param>
		public void Start(TimeSpan tasksUpdatePeriod)
		{
			this.tasksUpdatePeriod = tasksUpdatePeriod;
			using (ISession session = sessionFactory.OpenSession())
			{
				IList<TaskData> tasks = session.CreateCriteria<TaskData>().List<TaskData>();
				string[] disabled = (from t in tasks where !t.IsEnabled select t.ID).ToArray();
				string[] enabled = (from t in tasks where t.IsEnabled select t.ID).ToArray();
				logger.InfoFormat("{0} задач, запущены: [{1}], приостановлены: [{2}]",
					tasks.Count, string.Join(", ", enabled), string.Join(", ", disabled));
				processors = (from t in tasks
							  select new TaskProcessor(t, executor, schedule, stateManager, 
								  GetProcessingDelay()))
							  .ToDictionary(p => p.TaskID, p => p);
			}
			tasksUpdateTimer = new Timer(new TimerCallback(TimerCallback));
			tasksUpdateTimer.Change(tasksUpdatePeriod, tasksUpdatePeriod);
		}

		/// <summary>
		/// Остановить обработку задач
		/// </summary>
		/// <param name="waitBeforeStop">сколько нужно подождать для того чтобы все задачи успели завершиться</param>
		/// <returns>true - все задачи успели завершиться за отведенный период, false - некоторые не успели</returns>
		[MethodImplAttribute(MethodImplOptions.Synchronized)]
		public void Stop(TimeSpan waitBeforeStop)
		{
			tasksUpdateTimer.Change(TimeSpan.FromMilliseconds(-1), TimeSpan.FromMilliseconds(-1));
			IAsyncResult[] asyncStops = new IAsyncResult[processors.Count];
			// инициализируем остановку всех процессоров
			int i = 0;
			foreach (TaskProcessor p in processors.Values)
			{
				logger.DebugFormat("Начата остановка процессора задачи '{0}'.", p.TaskID);
				AsyncStop stopDelegate = new AsyncStop(p.StopGracefully);
				asyncStops[i++] = stopDelegate.BeginInvoke((int)waitBeforeStop.TotalMilliseconds, 
					new AsyncCallback(OnProcessorStop),
					p.TaskID);
			}
			// linq -- это круто.
			WaitHandle[] handles = asyncStops.Select(a => a.AsyncWaitHandle).ToArray();
			// тут мы не указываем время, поскольку его должны выдержать наши обработчики
			if (handles.Length > 0)
				WaitHandle.WaitAll(handles);
		}

		private void OnProcessorStop(IAsyncResult result)
		{
			logger.DebugFormat("Обработчик задачи '{0}' завершён.", result.AsyncState);
		}
		/// <summary>
		/// Делегат для асинхронного вызова всяческих остановок.
		/// </summary>
		/// <param name="timeoutMilliseconds">Время ожидания</param>
		protected delegate void AsyncStop(int timeoutMilliseconds);

		/// <summary>
		/// Callback-функция, чтобы не заворачивать содержимое UpdateTasks в уродливый try
		/// </summary>
		/// <param name="state"></param>
		private void TimerCallback(object state)
		{
			try
			{
				UpdateTasks();
			}
			catch (Exception e)
			{
				logger.Error("Ошибка обновления списка задач", e);
			}
		}
		/// <summary>
		/// Обновить состояние задач
		/// </summary>
		private void UpdateTasks()
		{
			using (ISession session = sessionFactory.OpenSession())
			{
				logger.DebugFormat("Обработка состояния задач.");
				// 1) убить все задачи, которые запущены, но отменены
				List<TaskProcessor> aborted = new List<TaskProcessor>();
				// список запущенных выполнений
				List<int> running = new List<int>();
				foreach (TaskProcessor p in processors.Values)
				{
					// это опять же на случай изменения в другом потоке
					int? execution = p.ExecutionID;
					if (execution.HasValue)
						running.Add(execution.Value);
				}
				if (running.Count > 0)
				{
					IList<TaskExecution> cancelled = session.CreateCriteria<TaskExecution>()
						.Add(Expression.In(TaskExecution.IDProp, running.ToArray()))
						.Add(Expression.Eq(TaskExecution.StateProp, ExecutionState.Cancelled))
						.List<TaskExecution>();
					foreach (TaskExecution e in cancelled)
					{
						TaskProcessor p = processors[e.Task.ID];
						if (p.Abort(e.ID))
						{
							logger.DebugFormat("Выполнение задачи '{0}' с идентификатором {1} отменено.",
								e.Task.ID, e.ID);
							// 
							processors.Remove(e.Task.ID);
							aborted.Add(p);
						}
					}
				}
				// если кого-то долго не обновляли, его надо или обновить или убить
				IList<TaskExecution> list = session.CreateCriteria<TaskExecution>()
					.Add(Expression.Eq(TaskExecution.StateProp, ExecutionState.Running))
					.Add(Expression.Lt(TaskExecution.UpdatedAtProp, DateTime.Now - tasksUpdatePeriod))
					.List<TaskExecution>();
				foreach (TaskExecution te in list)
				{
					// это случай, когда задачу долго не обновляли, а выполняем не мы
					if (!running.Contains(te.ID))
					{
						logger.InfoFormat("Обнаружена зависшая задача {0}[{1}]. Запущена в {2}, последнее обновление в {3}.",
							te.Task.ID, te.ID, te.StartTime, te.UpdatedAt);
						// переводим в запущенное состояние, пусть разбирается запускатор
						te.State = ExecutionState.Created;
					}
					te.UpdatedAt = DateTime.Now;
					session.Update(te);
					session.Flush();
				}

				Dictionary<string, TaskData> tasks = session.CreateCriteria<TaskData>()
					.List<TaskData>()
					.ToDictionary(a => a.ID, a => a);
				// обновим список задач
				foreach (TaskProcessor p in processors.Values)
				{
					if (!tasks.ContainsKey(p.TaskID))
					{
						// процессор для задачи удалён
						p.Abort();
						processors.Remove(p.TaskID);
						logger.InfoFormat("Задача {0} удалена из списка запусков.");
					}
					else
					{
						// не надо ли обновить процессор для задачи
						p.UpdateTask(tasks[p.TaskID]);
						// в словаре должны остаться задачи, для которых нет процессора
						tasks.Remove(p.TaskID);
					}
				}
				// если остались задачи, то для них не было процессоров и их надо запустить
				foreach (TaskData task in tasks.Values)
				{
					session.Evict(task);
					logger.InfoFormat("Запущена задача {0}", task.ID);
					TaskProcessor p = new TaskProcessor(task, executor, schedule,
						stateManager, GetProcessingDelay());
					processors.Add(task.ID, p);
				}
				// 3) запустить задачи, если есть ожидающие запуски
				IList<TaskData> pending = session.CreateCriteria<TaskExecution>()
					.Add(Expression.Le(TaskExecution.StartTimeProp, DateTime.Now))
					.Add(Expression.Eq(TaskExecution.StateProp, ExecutionState.Created))
					.SetProjection(Projections.Distinct(
						Projections.Property(TaskExecution.TaskProp)))
					.List<TaskData>();
				foreach (TaskData pendingTask in pending)
					processors[pendingTask.ID].TriggerManual();
			}
		}

		/// <summary>
		/// Уведомить диспетчер об изменении списка задач
		/// </summary>
		public void NotifyTasksUpdated()
		{
			tasksUpdateTimer.Change(TimeSpan.FromSeconds(0), tasksUpdatePeriod);
		}

		/// <summary>
		/// Узнать, насколько задерживать запуск обработчика
		/// </summary>
		/// <returns></returns>
		private int GetProcessingDelay()
		{
			return processingDelaySeconds == 0 ? 0 : generator.Next(processingDelaySeconds);
		}
	}
}
